package com.afdor.rws.event.channel.client.consumer;

import com.afdor.rws.core.extension.ExtensionLoader;
import com.afdor.rws.event.Event;
import com.afdor.rws.event.Serialization;
import com.afdor.rws.event.channel.client.consumer.handler.EventHandler;
import com.afdor.rws.event.channel.client.consumer.handler.EventRetryStrategy;
import com.afdor.rws.event.channel.filter.EventInterceptChain;
import com.afdor.rws.event.config.EventConfigUtils;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.NonNull;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;

import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

/**
 * 消费者抽象实现
 *
 * @author 悭梵
 * @date Created in 2018-08-15 10:40
 */
@ToString
@EqualsAndHashCode
public abstract class AbstractEventConsumer implements EventConsumer, Runnable {
    private volatile boolean init = Boolean.FALSE;
    @Getter
    @NonNull
    private volatile String eventSourceName;
    /**
     * 事件处理
     */
    @Getter
    @NonNull
    private EventHandler eventHandler;
    /**
     * 消费失败重试策略
     */
    @Getter
    @NonNull
    private EventRetryStrategy eventRetryStrategy;
    /**
     * 序列化方式
     */
    @Getter
    @NonNull
    private volatile Serialization serialization;
    @NonNull
    private volatile EventInterceptChain interceptChain;

    /**
     * 消费者线程池
     */
    protected static final ScheduledThreadPoolExecutor EVENT_CONSUMER_THREAD_POOL = new ScheduledThreadPoolExecutor(5, new ThreadFactory() {
        /**
         * 线程计数器
         */
        private AtomicInteger threadIndex = new AtomicInteger(0);

        /**
         * 线程组
         */
        private ThreadGroup threadGroup = new ThreadGroup("event-consumer") {
            {
                setDaemon(true);
                setMaxPriority(Thread.NORM_PRIORITY);
            }
        };

        private Thread.UncaughtExceptionHandler uncaughtExceptionHandler = new EventConsumerUncaughtExceptionHandler();

        @Override
        public Thread newThread(Runnable runnable) {
            Thread thread = new Thread(threadGroup, runnable, "thread-event-consumer-" + threadIndex.getAndIncrement());
            thread.setDaemon(true);
            thread.setPriority(Thread.MIN_PRIORITY);
            thread.setUncaughtExceptionHandler(uncaughtExceptionHandler);
            return thread;
        }
    });

    @Override
    public AbstractEventConsumer setEventSourceName(@NonNull String eventSourceName) {
        if (StringUtils.isNotBlank(this.eventSourceName)) {
            throw new RuntimeException("事件源不支持修改");
        }
        this.eventSourceName = eventSourceName;

        return this;
    }

    @Override
    public AbstractEventConsumer setEventHandler(@NonNull final EventHandler eventHandler) {
        if (eventHandler != null) {
            this.eventHandler = new ProxyEventHandler(eventHandler);
        }
        return this;
    }

    @Override
    public AbstractEventConsumer setEventRetryStrategy(EventRetryStrategy eventRetryStrategy) {
        this.eventRetryStrategy = eventRetryStrategy;
        return this;
    }

    @Override
    public AbstractEventConsumer setSerialization(@NonNull Serialization serialization) {
        this.serialization = serialization;
        return this;
    }

    @Override
    public AbstractEventConsumer init() {
        if (interceptChain == null) {
            synchronized (this) {
                if (interceptChain == null) {
                    interceptChain = ExtensionLoader
                            .getInstance(EventInterceptChain.class)
                            .getExtensionOrDefault(EventConfigUtils
                                    .getProperty(EventConfigUtils.EVENT_CONSUMER_INTERCEPT_CHAIN_NAME, "default"));
                    init = Boolean.TRUE;
                }
            }
        }
        return this;
    }

    @Override
    public void start() {
        doStart();
        int period = NumberUtils.createInteger(EventConfigUtils.getProperty(EventConfigUtils.EVENT_CONSUMER_HANDLER_PERIOD, "1000"));
        if (period <= 0) {
            period = 1000;
        }
        // scheduleAtFixedRate  在initialDelay时长后第一次执行任务，以后每隔period时长，再次执行任务。注意，period是从任务开始执行算起的。开始执行任务后，定时器每隔period时长检查该任务是否完成，如果完成则再次启动任务，否则等该任务结束后才再次启动任务
        // scheduleWithFixedDelay 在initialDelay时长后第一次执行任务，以后每当任务执行完成后，等待delay时长，再次执行任务。
        // 两个方法均在initialDelay时长后第一次执行任务，以后每隔period时长，再次执行任务，但scheduleAtFixedRate以开始时间计算间隔，而scheduleWithFixedDelay以结束时间计算间隔
        EVENT_CONSUMER_THREAD_POOL.scheduleWithFixedDelay(this, 0, period, TimeUnit.MILLISECONDS);
    }

    @Override
    public void stop() {
        doStop();
        EVENT_CONSUMER_THREAD_POOL.shutdown();
    }

    @Override
    public void run() {
        if (!Thread.currentThread().isInterrupted() && hasEvent()) {
            /* 线程未终止，就执行事件消费。若处理事件返回为true，这继续循环处理事件； 若处理事件返回为false，则退出执行，有计划任务重新启动执行*/
            while (consume()) {
                if (Thread.currentThread().isInterrupted()) {
                    break;
                }
            }
        }
    }

    /**
     * 启动
     */
    protected void doStart() {

    }

    /**
     * 停止
     */
    protected void doStop() {

    }

    /**
     * 是否存在待消费消息
     *
     * @return
     */
    protected boolean hasEvent() {
        /**
         * 永远有消息
         */
        return true;
    }

    /**
     * 消费具体消息
     *
     * @return 返回true表示消费到事件，可能还有更多事件待消费；false表示没有消费到事件
     */
    protected abstract boolean consume();

    /**
     * 验证初始话是否成功
     */
    private void checkup() {
        int waitCount = 0;
        while (!this.init || this.interceptChain == null || waitCount == 100) {
            LockSupport.parkNanos(TimeUnit.MICROSECONDS.toNanos(10));
            waitCount++;
        }
        if (!this.init || this.interceptChain == null) {
            throw new IllegalStateException("please execute the init method first.");
        }
    }

    /**
     * 代理事件处理器对象
     */
    public class ProxyEventHandler implements EventHandler {
        /**
         * 原始事件处理器对象
         */
        private final EventHandler targetEventHandler;

        public ProxyEventHandler(EventHandler eventHandler) {
            this.targetEventHandler = eventHandler;
        }

        @Override
        public void handler(Event event) {
            // 拦截策略
            AbstractEventConsumer.this.checkup();
            EventInterceptChain interceptChain = AbstractEventConsumer.this.interceptChain.copy();
            interceptChain.doIntercept(event);
            // 未被拦截，在处理事件
            if (!interceptChain.isIntercept()) {
                this.targetEventHandler.handler(event);
            }
        }
    }

    /**
     * 线程异常拦截输出
     */
    @Slf4j
    public static class EventConsumerUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {

        @Override
        public void uncaughtException(Thread t, Throwable e) {
            log.error("thread " + t.getName() + " exception", e);
        }
    }
}
